package org.project.appmark

import org.FlinkStreamApp
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.project.bean.SimulatedEventSource

/**
 * description ：APP分渠道数据统计
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/27 11:39
 * modified By ：
 * version:    : 1.0
 */
object AppMarketingByChannel extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // 获取源数据
    val source = environment.addSource(new SimulatedEventSource)

    // app 分渠道统计
    source
      .assignAscendingTimestamps(_.ts * 1000)
      .map(ss => ((ss.channel, ss.behavior), 1L))
      .keyBy(_._1)
      .timeWindow(Time.seconds(60), Time.seconds(5))
      .process(new MarketingByChannel)
      .print()
  }

  class MarketingByChannel extends ProcessWindowFunction[((String, String), Long), ((String, String), Long), (String, String), TimeWindow] {
    override def process(key: (String, String), context: Context, elements: Iterable[((String, String), Long)], out: Collector[((String, String), Long)]) = {
      out.collect((key, elements.size))
    }
  }

}
